package com.atguigu.state;

/*
* 状态的TTL - Time To Live
*
* 给状态设置过期时间，在指定的时间内，没有对状态进行操作（读或者写），状态会被清理
*
* 如果在指定时间内，对状态进行了操作，时间会重置。
* */

import com.atguigu.pojo.WaterSensor;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

public class Flink10_StateTTL {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //s1,100,1000(id,水为值:Integer，ts:Long)
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Integer.valueOf(fields[1].trim()), Long.valueOf(fields[2].trim()));
                        }
                );
        ds.print("INPUT");

        //需求：针对每种传感器输出最高的3个水位值

        ds
                .keyBy(WaterSensor::getId)
                .process(

                        new KeyedProcessFunction<String, WaterSensor, String>() {

                            //1.声明状态
                            private ListState<Integer> Top3VcState;
                            //2.使用生命周期open方法，初始化状态

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                RuntimeContext runtimeContext = getRuntimeContext();
                                ListStateDescriptor<Integer> top3VcStateDes = new ListStateDescriptor<>("Top3VcStateDes", Types.INT);
                                //TTL ，设置过期时间
                                StateTtlConfig stateTtlConfig = //todo 创建一个状态生存时间配置对象，
                                        StateTtlConfig.newBuilder(Time.seconds(10)) //设置状态的生存时间为10秒。
                                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)//设置状态的更新类型为在读取和写入时更新。
                                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)//)设置状态的可见性为从不返回已过期的状态。
                                        .build();
                                top3VcStateDes.enableTimeToLive(stateTtlConfig);//这个状态可以存活时间

                                Top3VcState = runtimeContext.getListState(top3VcStateDes);//初始化后状态传过去
                            }

                            @Override
                            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context context, Collector<String> out) throws Exception {
                            //3.使用状态
                            //将当前的水位值添加到状态中
                                Top3VcState.add(value.getVc());
                            //取出状态的值
                                ArrayList<Integer>  Top3VcList = new ArrayList<>();
                                Top3VcState.get().forEach(top3VcList -> Top3VcList.add(top3VcList));
                                Top3VcList.sort(
                                        new Comparator<Integer>() {
                                            @Override
                                            public int compare(Integer o1, Integer o2) {
                                                return Integer.compare(o2,o1);//倒序是o2在前面，o1在后面
                                            }
                                        }
                                );
                                if (Top3VcList.size()>3){
                                    Top3VcList.remove(3);
                                }
                                //更新状态
                                out.collect("传感器：" + context.getCurrentKey() + ",最高的三个水位：" + Top3VcList);
                            }

                        }
                )
                .print("PROCESS");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
